package com.ds.lens.flink.monitor.monitorjob;

import com.ds.lens.data.common.document.dto.MonitorEvent;
import com.ds.lens.flink.monitor.MonitorJobContext;
import com.ds.lens.flink.monitor.common.CommonSerializationSchema;
import com.ds.lens.flink.monitor.common.KafkaKeyedPartition;
import com.ds.lens.flink.monitor.common.LensElasticsearchSink;
import com.ds.lens.flink.monitor.common.MonitorEventSerializeSchema;
import com.ds.lens.flink.monitor.common.bo.MetricBundle;
import com.ds.lens.flink.monitor.forecast.ForecastConfig;
import com.ds.lens.flink.monitor.forecast.function.ForecastActualDataFunction;
import com.ds.lens.flink.monitor.forecast.function.ForecastFilter;
import com.ds.lens.flink.monitor.monitorjob.function.*;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.http.HttpHost;

import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class LensMonitorJob {
    public static void main(String[] args) throws Exception {
        Properties properties = MonitorJobContext.getInstance().getKafkaProperties();
        if (properties == null) {
            System.err.println("No kafka properties found");
            return;
        }
        String outputTopic = MonitorJobContext.getInstance().getMonitorEventTopic();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
        env.getConfig().setParallelism(MonitorJobContext.getInstance().getJobParallelism());
        ForecastConfig forecastConfig = MonitorJobContext.getInstance().getForecastConfig();
        List<HttpHost> transportAddresses = MonitorJobContext.getInstance().getEsTransportAddresses();
        Map<String, String> esConfig = MonitorJobContext.getInstance().getEsConfig();

        /**
         * 绑定数据
         */
        DataStream<MetricBundle> dataStream = env.addSource(new LensMonitorSource(10, TimeUnit.SECONDS))
                .keyBy(new MonitorKeyKeySelector())
                .timeWindow(Time.minutes(1))
                .process(new MetricRetriveFunction()).name("MetricRetrieveFunction");

        /**
         * 发送实际指标数据给预测平台
         */
        dataStream.filter(new ForecastFilter()).name("ForecastFilter")
                .process(new ForecastActualDataFunction()).name("ForecastActualDataFunction")
                .addSink(new FlinkKafkaProducer<>(forecastConfig.getForecastActualDataTopic(), new CommonSerializationSchema<>(),
                        forecastConfig.getKafkaProperties(), Optional.of(new KafkaKeyedPartition<>()))).name("sinkActual");

        /**
         * 异常检测
         */
        DataStream<Tuple2<MetricBundle, Map<String, MonitorEvent>>> eventDataStream = dataStream.keyBy(new MetricKeySelector())
                .process(new MonitorEventGenerateFunction()).name("MonitorEventGenerateFunction");
        DataStream<MonitorEvent> eventDataResultStream = eventDataStream.process(new UpdateEventStateFunction()).name("UpdateEventStateFunction");

        /**
         * 发送检测event给下游job
         */
        eventDataResultStream.addSink(new FlinkKafkaProducer<>(outputTopic, new MonitorEventSerializeSchema(), properties, Optional.of(new KafkaKeyedPartition<>()))).name("sinkEvent");

        /**
         * 将动态检测实际值、预测值、检测结果等信息写入es
         */
        eventDataStream.filter(new DynamicDetectionFilter()).name("DynamicDetectionFilter")
                .process(new DynamicDetectionResultFunction()).name("DynamicDetectionResultFunction")
                .addSink(new LensElasticsearchSink<>(transportAddresses, new DynamicDetectionResultSinkFunction(), esConfig).build()).name("DynamicDetectionResultSinkFunction");

        env.execute(LensMonitorJob.class.getSimpleName());
    }
}
